// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//
#include <pollux/vector/selectivity_vector.h>

#include <pollux/common/base/nulls.h>

namespace kumo::pollux {
    // static
    const SelectivityVector &SelectivityVector::empty() {
        static SelectivityVector kEmpty{SelectivityVector(0, false)};
        return kEmpty;
    }

    SelectivityVector SelectivityVector::empty(vector_size_t size) {
        return SelectivityVector{size, false};
    }

    std::string SelectivityVector::toString(
        vector_size_t maxSelectedRowsToPrint) const {
        const auto selectedCnt = countSelected();

        POLLUX_CHECK_GE(maxSelectedRowsToPrint, 0);

        std::stringstream out;
        out << selectedCnt << " out of " << size() << " rows selected between "
                << begin() << " and " << end();

        if (selectedCnt > 0 && maxSelectedRowsToPrint > 0) {
            out << ": ";
            int cnt = 0;
            testSelected([&](auto row) {
                if (cnt > 0) {
                    out << ", ";
                }
                out << row;
                ++cnt;
                return cnt < maxSelectedRowsToPrint;
            });
        }
        return out.str();
    }

    void SelectivityVector::copy_nulls(uint64_t *dest, const uint64_t *src) const {
        if (isAllSelected()) {
            bits::copyBits(src, 0, dest, 0, size_);
            return;
        }

        const auto *rowBits = bits_.data();

        bits::forEachWord(
            begin_,
            end_,
            [dest, src, rowBits](int32_t idx, uint64_t mask) {
                // Set 'dest' to 0 for selected rows.
                dest[idx] = (dest[idx] & ~mask) | (mask & dest[idx] & ~rowBits[idx]);

                // Set 'copySrc' to 0 for non-selected rows.
                uint64_t copySrc =
                        (src[idx] & ~mask) | (mask & src[idx] & rowBits[idx]);

                // Combine 'dest' and 'copySrc' with an OR.
                dest[idx] = (dest[idx] & ~mask) | (mask & (dest[idx] | copySrc));
            },
            [dest, src, rowBits](int32_t idx) {
                // Set 'dest' to 0 for selected rows.
                dest[idx] = dest[idx] & ~rowBits[idx];

                // Set 'copySrc' to 0 for non-selected rows.
                uint64_t copySrc = src[idx] & rowBits[idx];

                // Combine 'dest' and 'copySrc' with an OR.
                dest[idx] = dest[idx] | copySrc;
            });
    }

    void translateToInnerRows(
        const SelectivityVector &outerRows,
        const vector_size_t *indices,
        const uint64_t *nulls,
        SelectivityVector &innerRows) {
        outerRows.applyToSelected([&](vector_size_t row) {
            if (!(nulls && bits::isBitNull(nulls, row))) {
                innerRows.setValid(indices[row], true);
            }
        });
        innerRows.updateBounds();
    }
} // namespace kumo::pollux
